
# 插入测试数据
import pymysql
import random

# 连接数据库函数
def connect_db(host, user, password, database, charset='utf8',port=3306)-> tuple[pymysql.Connection, pymysql.cursors.Cursor]:
    print(f'连接数据库 {host}...')
    '''
    host: 数据库地址
    user: 数据库用户名
    password: 数据库密码
    database: 数据库名称
    charset: 编码格式
    port: 端口号
    '''
    conn = pymysql.connect(host=host, user=user, password=password, database=database,charset=charset,port=port)
    cursor = conn.cursor()
    return conn, cursor
 
# 创建数据库函数
def create_database(conn, cursor, database_name):
    '''
    conn: 数据库连接
    cursor: 游标
    database_name: 数据库名称
    '''
    print(f'创建数据库 {database_name}...')
    cursor.execute(f'CREATE DATABASE IF NOT EXISTS {database_name}')
    print(f'数据库 {database_name} 创建成功！')
    conn.commit()

# 创建表函数
def create_table(conn, cursor, database_name, table_name):
    '''
    conn: 数据库连接
    cursor: 游标
    database_name: 数据库名称
    table_name: 表名称
    '''
    print(f'创建表 {table_name}...')
    cursor.execute(f'USE {database_name}')
    cursor.execute(f'CREATE TABLE IF NOT EXISTS {table_name} (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), age INT)')
    print(f'表 {table_name} 创建成功！')
# 插入测试数据函数
def insert_data(conn,cursor, database_name, table_name, num_records):
    '''
    conn: 数据库连接
    cursor: 游标
    database_name: 数据库名称
    table_name: 表名称
    num_records: 插入的记录数
    '''
    for i in range(num_records):
        name = 'user' + str(i)
        age = random.randint(18, 60)
        cursor.execute(f'INSERT INTO {table_name} (name, age) VALUES ("{name}", {age})')
    conn.commit()
 
    print('数据插入成功！')

#执行sql语句函数
def execute_sql(conn,cursor,sql):
    '''
    conn: 数据库连接
    cursor: 游标
    sql: sql语句
    '''
    print('执行SQL语句...')
    cursor.execute(sql)
    conn.commit()
 
#  删除数据库函数
def delete_database(conn, cursor, database_name):
    '''
    conn: 数据库连接
    cursor: 游标
    database_name: 数据库名称
    '''
    print(f'删除数据库 {database_name}...')
    cursor.execute(f'DROP DATABASE IF EXISTS {database_name}')
    print(f'数据库 {database_name} 删除成功！')
    conn.commit()
#  删除表函数
def delete_table(conn, cursor, database_name, table_name):
    '''
    conn: 数据库连接
    cursor: 游标
    database_name: 数据库名称
    table_name: 表名称
    '''
    print(f'删除表 {table_name}...')
    cursor.execute(f'DROP TABLE IF EXISTS {database_name}.{table_name}')
    print(f'表 {table_name} 删除成功！')
    conn.commit()
#  关闭数据库连接函数
def close_db(conn, cursor):
    '''
    conn: 数据库连接
    cursor: 游标
    '''
    print('关闭数据库连接...')
    cursor.close()
    conn.close()
    print('数据库连接已关闭！')

# 执行sql语句，读取数据
def execute_sql_and_read_data(conn, cursor, sql)-> list:
    '''
    conn: 数据库连接
    cursor: 游标
    sql: sql语句
    '''
    print('执行SQL语句...')
    cursor.execute(sql)
    result = cursor.fetchall()
    print('数据读取成功！')
    return result

import pandas as pd
# 导入SQLAlchemy的text函数
# text函数用于在SQLAlchemy中创建一个文本SQL表达式
# 这个表达式可以用于执行原始SQL查询，而不需要使用SQLAlchemy的ORM功能
from sqlalchemy import text
from sqlalchemy import create_engine
from urllib.parse import quote_plus

if __name__ == '__main__':
    
    host = 'localhost'
    user = 'root'
    password = 'Dydy@123dydy'
    database = 'test'
    port = 3306 
    sql="""
    SELECT * FROM t2;
    """
    # SELECT * FROM test.t2;

    print('使用 pymysql 读取数据')
    
    cursor=connect_db(host, user, password, database, charset='utf8',port=port)[1]
    cursor.execute(sql)
    result=cursor.fetchall()

   # 获取列名
    column_names = [desc[0] for desc in cursor.description]
    # 转换为 DataFrame
    df = pd.DataFrame(result, columns=column_names)    
    print(df.head())

    # print(df[:3])

    print('-'*55)
    print('使用 sqlalchemy 读取数据')

    # 对包含特殊字符的密码进行URL编码
    encoded_password = quote_plus(password)
    # 创建 SQLAlchemy 引擎    # 创建 SQLAlchemy 引擎
    engine = create_engine(f'mysql+pymysql://{user}:{encoded_password}@{host}:{port}/{database}?charset=utf8')
    # 使用 pandas 读取数据
    df = pd.read_sql(text(sql), con=engine)    
    print(df.head())


    exit()

